package com.fxz.rabbitmq.producer.broker;

import cn.hutool.core.util.ObjectUtil;
import com.fxz.rabbitmq.api.enums.MessageTypeEnum;
import com.fxz.rabbitmq.api.exception.MessageException;
import com.fxz.rabbitmq.api.pojo.Message;
import com.google.common.collect.Maps;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.retry.support.RetryTemplate;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;

import java.util.Map;

/**
 * 池化rabbitTemplate
 * 每一个topic对应一个RabbitTemplate
 * 1 提高发送的效率
 * 2 可以根据不同的需求定制化不同的RabbitTemplate
 * @author Fxz
 * @version 1.0
 * @date 2021-08-17 15:18
 */

@Slf4j
@Component
public class RabbitTemplateContainer implements RabbitTemplate.ConfirmCallback {

    @Autowired
    private ConnectionFactory connectionFactory;

    private Map<String, RabbitTemplate> rabbitMap = Maps.newConcurrentMap();

    public RabbitTemplate getTemplate(Message message) throws MessageException {
        if (ObjectUtil.isEmpty(message)) {
            throw new MessageException("message为空");
        }
        if (StringUtils.isEmpty(message.getTopic())) {
            throw new MessageException("topic为空!");
        }
        String topic = message.getTopic();
        RabbitTemplate rabbitTemplate = rabbitMap.get(topic);
        if (ObjectUtil.isNotEmpty(rabbitTemplate)) {
            return rabbitTemplate;
        }
        log.info("#RabbitTemplateContainer.getTemplate# topic:{} is not exists,create one#", topic);
        RabbitTemplate newRabbitTemplate = new RabbitTemplate(connectionFactory);
        newRabbitTemplate.setEncoding(topic);
        newRabbitTemplate.setRetryTemplate(new RetryTemplate());
        newRabbitTemplate.setRoutingKey(message.getRoutingKey());
        //对于message的序列化方式
        //newRabbitTemplate.setMessageConverter();

        String messageType = message.getMessageType();
        if (!MessageTypeEnum.RAPID.getCode().equals(messageType)) {
            newRabbitTemplate.setConfirmCallback(this);
        }
        rabbitMap.putIfAbsent(topic, newRabbitTemplate);
        return rabbitMap.get(topic);
    }

    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        //具体的消息应答
        String[] arr = correlationData.getId().split("#");
        String messageId = arr[0];
        Long sendTime = Long.valueOf(arr[1]);
        if (ack) {
            log.info("send message is ok,confirm messageId:{},sendTime:{}", messageId, sendTime);
        } else {
            log.error("send message is fail,confirm messageId:{},sendTime:{}", messageId, sendTime);
        }
    }
}
